package com.reload.stream.flow;

import java.io.IOException;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @Date: 2024/12/4 13:55
 * @Author: jianchenlong
 * @Description: 加上中间操作
 */
public class FlowProcessorDemo {
    public static void main(String[] args) throws IOException {
        SubmissionPublisher<String> submissionPublisher = new SubmissionPublisher<String>();
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + "-订阅开始了-" + subscription);
                this.subscription = subscription;
                // 请求1个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + "-订阅者接收到新数据-" + item);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + "-订阅者接收到错误-" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + "-生产者完成");

            }
        };

        // 定义一个中间操作，给每个元素加上 哈哈 前缀
        Flow.Processor<String, String> processor = new MyProcessor();

        // 订阅者绑定中间操作，中间操作绑定订阅者
        submissionPublisher.subscribe(processor);
        processor.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            // 发布者发布10条数据
            submissionPublisher.submit("submission - " + i);
        }

        System.in.read();
    }

    // 定义自己的中间处理器，只用写订阅者的接口
    static class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {
        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("MyProcessor 订阅绑定完成");
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(String item) {
            System.out.println("MyProcessor 接收到数据：" + item);
            // 对数据进行再加工
            submit("哈哈-" + item);
            subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {

        }
    }
}
